package xiaoa.java.push.queue;

import java.util.LinkedHashMap;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import xiaoa.java.log.L;
import xiaoa.java.utils.ExceptionUtils;


/**
 * queue 基础
 * @author lkc
 * @date 2017年10月17日 下午9:00:12
 * @version V1.0
 *
 */
public abstract class QueueBase<MsgBean> {
    
    // 线程池数量
    protected  BlockingQueue<Runnable> workQueue  = null;
    
    // 线程池
    protected   ThreadPoolExecutor  pool     = null;
    
    
    /**
     * 初始化
     * @Title: init
     * @param coreThread
     * @param maxThread
     * @author lkc
     */
    public void init(int coreThread , int maxThread){
        
        if (coreThread <= 0 || maxThread <=0 ){
             throw new RuntimeException("参数有误");
        }
        
        // 初始化队列
        workQueue =  new LinkedBlockingQueue<Runnable>(100000);
        
        // 初始化线程池
        pool      =  new ThreadPoolExecutor(coreThread, maxThread, 1L, TimeUnit.MINUTES, workQueue);
    }
    
    
    
    /**
     * 发送message
     * @Title: doPushMessage
     * @param bean
     * @author lkc
     */
    abstract void doPushMessage(MsgBean bean , LinkedHashMap<String, Object>  logMap )throws Throwable;
    
    /**
     * 推送消息
     * @Title: doPushMessage
     * @param bean
     * @param tid
     * @author lkc
     */
    public void pushMessage(final MsgBean  bean , final String tid){
        
        Runnable run = new Runnable() {
            
            @Override
            public void run() {
                
                LinkedHashMap<String, Object>   logMap = new LinkedHashMap<String, Object>();
                long startTime = System.currentTimeMillis();

                L.setTransactionIdThreadLocal(tid);
                try {
                    // 发送
                    doPushMessage(bean,logMap);
                    
                } catch (Throwable e) {
                   String message = ExceptionUtils.exceptionToString(e);
                   logMap.put("message", message);
                }
                logMap.put("queueSize", workQueue.size());
                L.showUsedTime("SendQueue", logMap , startTime , System.currentTimeMillis());
                
            }
        };
        
        pool.execute(run);
        
    }
    
    /**
     * 获取数量
     * 方法功能描述
     * 
     * @param
     * @since 2017年10月26日 下午5:47:33  
     * @author  virgo
     * @return
     */
    public  int size(){
        
        return workQueue.size();
        
    }
    

}
